package org.dromara.assets.service.impl;

import cn.hutool.core.collection.CollUtil;
import com.binance.connector.client.common.configuration.SignatureConfiguration;
import com.binance.connector.client.common.websocket.configuration.WebSocketClientConfiguration;
import com.binance.connector.client.common.websocket.dtos.StreamResponse;
import com.binance.connector.client.common.websocket.service.StreamBlockingQueueWrapper;
import com.binance.connector.client.spot.websocket.api.SpotWebSocketApiUtil;
import com.binance.connector.client.spot.websocket.api.api.SpotWebSocketApi;
import com.binance.connector.client.spot.websocket.api.model.ExecutionReport;
import com.binance.connector.client.spot.websocket.api.model.SessionSubscriptionsResponse;
import com.binance.connector.client.spot.websocket.api.model.UserDataStreamEventsResponse;
import com.binance.connector.client.spot.websocket.api.model.UserDataStreamSubscribeResponse;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.jetty.client.Socks5Proxy;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;

import static java.lang.Thread.sleep;

@Component
@Slf4j
@Scope("prototype") // 每次获取都是新实例
public class BinanceWebSocketWithNew {

    @Autowired
    BinanceWebSocketExecutionReportHandler executionReportHandler;

    public static SpotWebSocketApi api;

    private Thread consumerThread;

    public void start(String apiKey, String path, String pass) {
        getApi(apiKey, path, pass);
        // 启动消费者线程
        consumerThread = new Thread(this::userDataStreamSubscribeExampleStream);
        consumerThread.start();
    }

    // 重新连接方法
    private void reconnect() {
        log.info("Reconnecting Start");
        // 重新启动消费者线程
        consumerThread = new Thread(this::userDataStreamSubscribeExampleStream);
        consumerThread.start();
        log.info("Reconnecting End");
    }

    public static SpotWebSocketApi getApi(String apiKey, String path, String pass) {
        if (api == null) {
            WebSocketClientConfiguration clientConfiguration = SpotWebSocketApiUtil.getClientConfiguration();
            Socks5Proxy proxy = new Socks5Proxy("127.0.0.1", 1081);
            clientConfiguration.setWebSocketProxy(proxy);
            clientConfiguration.setAutoLogon(true);
            SignatureConfiguration signatureConfiguration = new SignatureConfiguration();
            signatureConfiguration.setApiKey(apiKey);
            signatureConfiguration.setPrivateKey(path);
            signatureConfiguration.setPrivateKeyPass(pass);
            clientConfiguration.setSignatureConfiguration(signatureConfiguration);
            api = new SpotWebSocketApi(clientConfiguration);
        }
        return api;
    }

    /**
     * WebSocket Subscribe to User Data Stream
     *
     * <p>Subscribe to the User Data Stream in the current WebSocket connection. Weight: 2
     */
    public void userDataStreamSubscribeExampleStream() {
        StreamResponse<UserDataStreamSubscribeResponse, UserDataStreamEventsResponse> resp = api.userDataStreamSubscribe();
        resp.getResponse().thenAccept(responseResult -> {
            log.info("订阅成功，{}", String.valueOf(responseResult));
        });
        StreamBlockingQueueWrapper<UserDataStreamEventsResponse> stream = resp.getStream();
        while (true) {
            try {
                UserDataStreamEventsResponse response = stream.take();
                if (response.getActualInstance() instanceof ExecutionReport) {
                    ExecutionReport executionReport = response.getExecutionReport();
                    executionReportHandler.hand(executionReport);
                } else {
                    log.info("收到其它消息推送:{}", response.getActualInstance().getClass().getName());
                }
            } catch (InterruptedException e) {
                log.warn("线程被中断");
                break;
            } catch (Exception e) {
                log.error(e.getMessage(), e);
                try {
                    sleep(10000);
                } catch (InterruptedException e1) {
                    log.error("", e1);
                }

            }
        }
    }

    @Scheduled(fixedRate = 5 * 60 * 1000) // 每隔5分钟执行一次
    public void executeTask() throws Exception {
        CompletableFuture<SessionSubscriptionsResponse> resp = api.sessionSubscriptions();
        log.info("sessionSubscriptions: {}", resp.get());
        if (CollUtil.isEmpty(resp.get().getResult())) {
            log.warn("未查询到订阅信息，中断线程重新订阅");
            consumerThread.interrupt();
            reconnect();
        }
    }
}
